package feizhou.untils.tools;

import com.sun.istack.NotNull;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * author harry
 * email:xianheng.huang@lolaage.com
 * createDate:2018/5/14
 * 该类主要是用来执行分片任务，需要返回列表的值的工具，使用者可以自己指定线程池，也可以使用默认线程池，是一个多实例类，方便拓展
 */
public class FutureThreadPoolUtil {
    private Logger log = Logger.getLogger(FutureThreadPoolUtil.class);
    private ExecutorService executorService;

    private ConcurrentHashMap<Callable, Future> callabeMap;

    public void execute(Runnable runnable) {
        executorService.execute(runnable);
    }

    /**
     * 构造方法中实例化线程池
     *
     * @param threadName 指定构造方法线程名称，方便如果出错时排查错误
     */
    public FutureThreadPoolUtil(String threadName) {
        this(Runtime.getRuntime().availableProcessors() + 1, threadName);
    }

    public FutureThreadPoolUtil(int nThreads, String threadName) {
        if (nThreads <= 0) {
            return;
        }
        if (StringUtil.isEmpty(threadName)) {
            executorService = Executors.newFixedThreadPool(nThreads);
        } else {
            executorService = Executors.newFixedThreadPool(nThreads, new ThreadFactory() {
                private AtomicInteger count = new AtomicInteger();

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, threadName + "-" + count.incrementAndGet());
                }
            });
        }
    }

    /**
     * 通过使用者传入线程池的方式创建线程池，使用该方式主要是拓展线程池的使用和管理，初始化线程池方式适用于短暂的执行任务，
     * 而对于其他任务的线程管理，还是通过传入的方式管理更好
     *
     * @param executorService
     */
    public void createExecutorService(@NotNull ExecutorService executorService) {
        this.executorService = executorService;
    }

    /**
     * 根据指定的callable集合，采用completionService结合线程池来执行任务，
     * 任务的耗时操作时间不确定，有快有慢，且不需要排序的，先执行完成的线程，重新回来拿任务，再次去执行
     *
     * @param callables
     * @param <T>       需要返回的泛型类型
     * @return 返回指定的类型集合，无序
     */
    public <T> List<T> submit(List<Callable<T>> callables) {
        if (callables == null || callables.isEmpty()) {
            return null;
        }
        CompletionService completionService = new ExecutorCompletionService(executorService);
        for (Callable<T> callable : callables) {
            completionService.submit(callable);
        }

        List<T> list = new ArrayList<>();
        try {
            for (Callable<T> callable : callables) {
                Future<T> take = completionService.take();
                list.add(take.get());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return list;
    }

    /**
     * 根据指定集合callable去执行任务，任务执行完后统一取出，
     *
     * @param callables
     * @param <T>
     * @return
     */
    public <T> List<T> submitSort(List<Callable<T>> callables) {
        return submitSort(callables, "default");
    }

    /**
     * 根据指定集合callable去执行任务，任务执行完后统一取出，
     *
     * @param callables
     * @param <T>
     * @return
     */
    public <T> List<T> submitSort(List<Callable<T>> callables, String actionLog) {
        if (callables == null || callables.isEmpty()) {
            return null;
        }
        long startTime = System.currentTimeMillis();
        BlockingQueue<Future<T>> queue = new LinkedBlockingDeque<>();
        for (Callable<T> callable : callables) {
            Future<T> submit = executorService.submit(callable);
            queue.add(submit);
        }

        List<T> list = new ArrayList<>();
        try {
            int size = callables.size();
            for (int x = 0; x < size; x++) {
                Future<T> take = queue.take();
                T t = take.get();
                if (t != null) {
                    list.add(t);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        long endTime = System.currentTimeMillis();
        long useTime = endTime - startTime;
        if (useTime > 1000) {
            log.error(actionLog + "指定集合callable去执行任务,总耗时=" + (endTime - startTime) + "耗秒,list.size=" + list.size() + "," +
                    "callables.size=" + callables.size());
        }
        if (list.size() != callables.size()) {
            log.error(actionLog + "指定集合callable任务结果不等,总耗时=" + (endTime - startTime) + "耗秒,list.size=" + list.size() + "," +
                    "callables.size=" + callables.size());
        }
        return list;
    }

    private ConcurrentHashMap getCallabeMap() {
        if (callabeMap == null) {
            synchronized (FutureThreadPoolUtil.class) {
                if (callabeMap == null) {
                    callabeMap = new ConcurrentHashMap();
                }
            }
        }
        return callabeMap;
    }

    /**
     * 将callable使用线程执行，提交后将future存放到map集合，最后在通过调用{@see takeResult(callable)}获取任务执行的结果
     *
     * @param callable
     * @param <T>
     */
    public <T> void submitATask(Callable<T> callable) {
        if (callable == null) {
            return;
        }
        Future<T> submit = executorService.submit(callable);
        ConcurrentHashMap callabeMap = getCallabeMap();
        callabeMap.putIfAbsent(callable, submit);
    }

    /**
     * 调用该方法可以获取执行任务的返回结果，调用之前，需要先调用{@see submitATask(callable)},将callable传给线程执行
     *
     * @param callable
     * @param <T>
     * @return 返回{@see submitATask(callable)}中callable传入的泛型类型
     */
    public <T> T takeResult(Callable<T> callable) {
        Future remove = (Future) getCallabeMap().remove(callable);
        try {
            return (T) remove.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return null;
    }

}
